package com.tuijian.util;

import com.tuijian.vo.CellData;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.*;
@Component
public class HbaseUtils {
    private static final Logger log=Logger.getLogger(HbaseUtils.class);
    @Value("${spring.hbase.zkHost}")
    private String zkHost;
    @Value("${spring.hbase.zkPort}")
    private String zkPort;
    private Connection conn;
    /**
     *
     * @param zkAddr zookeeper集群地址，多个用 , 分割
     * @param port zookeeper端口号，默认是2181
     * @throws IOException
     */
    public void connect(String zkAddr,String port) throws IOException {
        //创建hbase配置对象
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zkHost);// zookeeper地址
        conf.set("hbase.zookeeper.property.clientPort", zkPort);// zookeeper端口
        //通过配置对象创建连接
        conn = ConnectionFactory.createConnection(conf);
    }
    /**
     * 关闭连接
     * @throws IOException
     */
    public void close() throws IOException {
        conn.close();
    }

    /**
     * 创建表
     * @param tableName 表名
     * @param columnFamily 列族名
     * @return true：表示创建成功；false：表示表已存在
     * @throws IOException
     */
    public boolean createTable(String tableName,String columnFamily) throws IOException{
        boolean ret = false;
        TableName tableNameObj = TableName.valueOf(tableName);
        // 创建admin对象，判断表是否存在，创建表、禁用表、删除表
        Admin admin = conn.getAdmin();
        if(admin.tableExists(tableNameObj)){
            ret = false;
        }else {
            //创建表描述构造器，创建表时，可以设置很多配置
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            //创建列族描述构造器，创建表时，可以设置很多配置
            //在hbase中，数据是以 byte[] 存储的，所以需要做转换
            ColumnFamilyDescriptorBuilder cfdb = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily));
            //创建列族描述对象
            ColumnFamilyDescriptor cfd = cfdb.build();
            //将列族描述对象传入表描述构造器
            tableDescriptorBuilder.setColumnFamily(cfd);
            //创建表描述对象
            TableDescriptor td = tableDescriptorBuilder.build();
            //创建表
            admin.createTable(td);
            ret = true;
        }
        admin.close();
        return ret;
    }

    /**
     * 删除表
     * @param tableName 表名
     * @return true：表示删除成功；false：表示表不存在
     * @throws IOException
     */
    public boolean dropTable(String tableName) throws IOException {
        boolean ret=false;
        Admin admin=conn.getAdmin();
        TableName tableNameObj = TableName.valueOf(tableName);
        //判断表是否存在
        if(admin.tableExists(tableNameObj)) {
            //禁用表
            admin.disableTable(tableNameObj);
            //删除表
            admin.deleteTable(tableNameObj);
            ret=true;
        }else{
            ret=false;
        }
        return ret;
    }

    /**
     * 添加数据，可以添加多个列、值
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param columnValue
     * @throws IOException
     */
    public void put(String tableName,String rowKey,String columnFamily,Map<String,String> columnValue) throws IOException {
        //获取table对象
        Table table = conn.getTable(TableName.valueOf(tableName));
        //创建put对象，对应put命令，封装rowkey、列族、列名、列值
        Put put = new Put(Bytes.toBytes(rowKey));
        Set<Map.Entry<String, String>> set = columnValue.entrySet();
        for(Map.Entry<String, String> entry : set) {
            String column=entry.getKey();
            String value=entry.getValue();
            put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        }
        //执行put命令
        table.put(put);
        //关闭table
        table.close();
    }

    /**
     * get指令
     * @param tableName
     * @param rowKey
     * @param columnFamily，查询指定列族，为null表示全查
     * @param qualifier，查询指定列，为null表示全查
     * @return
     * @throws IOException
     */
    public List<CellData> get(String tableName,String rowKey,String columnFamily,String qualifier) throws IOException {
        //获取table对象
        Table table = conn.getTable(TableName.valueOf(tableName));
        //根据rowkey构建Get对象
        Get get = new Get(Bytes.toBytes(rowKey));
        if(qualifier!=null && !"".equals(qualifier)){
            // 如果 qualifier 不为空，那么就读取 columnFamily 列族中 qualifier 列的数据
            get.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(qualifier));
        }
        if (columnFamily != null && !"".equals(columnFamily) && ( qualifier==null || "".equals(qualifier))) {
            // 如果 columnFamily 不为空，qualifier为空，那么就读取 columnFamily 列族中的所有数据
            get.addFamily(Bytes.toBytes(columnFamily));
        }
        //执行get方法
        Result result=table.get(get);
        //获取rowkey
        String rowkey=Bytes.toString(result.getRow());
        //封装返回结果
        List<CellData> retList=new ArrayList<>();
        //获取所有单元格
        Cell[] resultList=result.rawCells();
        if(resultList!=null) {
            for (Cell item : resultList) {
                String family = Bytes.toString(CellUtil.cloneFamily(item));
                String qualifierOfDb = Bytes.toString(CellUtil.cloneQualifier(item));
                String value = Bytes.toString(CellUtil.cloneValue(item));
                String row = Bytes.toString(CellUtil.cloneRow(item));
                long timestamp = item.getTimestamp();
                CellData data = new CellData(family, qualifierOfDb, rowkey, timestamp, value);
                retList.add(data);
            }
        }
        //关闭
        table.close();
        return retList;
    }

    /**
     * 删除单元格数据
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @throws IOException
     */
    public void deleteCell(String tableName,String rowKey,String columnFamily,String column) throws IOException {
        TableName tn = TableName.valueOf(tableName);
        Table table = conn.getTable(tn);
        Delete delete=new Delete(Bytes.toBytes(rowKey));
        delete.addColumn(Bytes.toBytes(columnFamily),Bytes.toBytes(column));
        table.delete(delete);
        table.close();
    }

    /**
     * 根据rowKey，删除某行的数据
     * @param tableName
     * @param rowKey
     * @throws IOException
     */
    public void delete(String tableName,String rowKey) throws IOException {
        TableName tn = TableName.valueOf(tableName);
        Table table = conn.getTable(tn);
        Delete delete=new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
        table.close();
    }

    /**
     * 根据rowKey，删除某列族的数据
     * @param tableName
     * @param rowKey
     * @param family
     * @throws IOException
     */
    public void deleteFamily(String tableName,String rowKey,String family) throws IOException {
        TableName tn = TableName.valueOf(tableName);
        Table table = conn.getTable(tn);
        Delete delete=new Delete(Bytes.toBytes(rowKey));
        delete.addFamily(Bytes.toBytes(family));
        table.delete(delete);
        table.close();
    }

    /**
     *
     * @param tableName
     * @param family，返回指定的 family，指定null表示返回所有
     * @param column，返回指定的 family、column，指定null表示返回所有
     * @param startRow，指定开始的rowkey（包含）。如果不设置，则从表头开始
     * @param stopRow，指定结束的rowkey（不包含）。如果不设置，则截止的末尾
     * @return
     * @throws IOException
     */
    public List<CellData[]> scan(String tableName, String family, String column,String startRow,String stopRow) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        //创建Scan
        Scan scan=new Scan();
        // 返回指定的 family、column，不指定表示返回所有
        if(column!=null && !"".equals(column)) {
            scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
        }
        // 返回指定的 family，不指定表示返回所有
        if(family!=null && !"".equals(family) && (column==null || "".equals(column))) {
            scan.addFamily(Bytes.toBytes(family));
        }
        // 指定开始的行（包含）。如果不设置，则从表头开始
        if(startRow!=null && !"".equals(startRow)) {
            scan.withStartRow(Bytes.toBytes(startRow));
        }
        // 指定结束的行（不包含）。如果不设置，则截止的末尾
        if(stopRow!=null && !"".equals(stopRow)) {
            scan.withStopRow(Bytes.toBytes(stopRow));
        }
        //执行scan
        ResultScanner resultScanner = table.getScanner(scan);
        Iterator<Result> iterator = resultScanner.iterator();
        List<CellData[]> retList = new ArrayList<>();
        //迭代，在迭代时，ResultScanner会发送get请求
        while(iterator.hasNext()){
            Result result=iterator.next();
            //获取rowkey
            String rowkey=Bytes.toString(result.getRow());
//            System.out.print(rowkey+"\t");
            //获取所有单元格
            List<Cell> list=result.listCells();
            CellData[] arr = new CellData[list.size()];
            int i = 0;
            for(Cell item:list){
                String familyOfDb=Bytes.toString(CellUtil.cloneFamily(item));
                String qualifier=Bytes.toString(CellUtil.cloneQualifier(item));
                String valueDb=Bytes.toString(CellUtil.cloneValue(item));
                CellData cellData = new CellData(familyOfDb,qualifier,rowkey,item.getTimestamp(),valueDb);
                arr[i] = cellData ;
                i ++ ;
            }
            retList.add(arr);
        }
        //用完要及时关闭，在迭代时，ResultScanner会发送get请求
        resultScanner.close();
        table.close();
        return retList;
    }

    /**
     *
     * @param tableName 表名
     * @param family 列族名
     * @param column 列名
     * @param startRow，指定开始的rowkey（包含）。如果不设置，则从表头开始
     * @param stopRow，指定结束的rowkey（不包含）。如果不设置，则截止的末尾
     * @param op 操作，=，>，>=，<，<=，!=
     * @param rowComparator 比较器
     * @throws IOException
     */
    public List<CellData[]> filter(
            String tableName,
            String family,
            String column,
            String startRow,
            String stopRow,
            String op,
            ByteArrayComparable rowComparator) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        //创建Scan
        Scan scan=new Scan();
        // 返回指定的 family、column，不指定表示返回所有
        if(column!=null && !"".equals(column)) {
            scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(column));
        }
        // 返回指定的 family，不指定表示返回所有
        if(family!=null && !"".equals(family) && (column==null || "".equals(column))) {
            scan.addFamily(Bytes.toBytes(family));
        }
        // 指定开始的行（包含）。如果不设置，则从表头开始
        if(startRow!=null && !"".equals(startRow)) {
            scan.withStartRow(Bytes.toBytes(startRow));
        }
        // 指定结束的行（不包含）。如果不设置，则截止的末尾
        if(stopRow!=null && !"".equals(stopRow)) {
            scan.withStopRow(Bytes.toBytes(stopRow));
        }
        RowFilter filter = new RowFilter(getCompareOperator(op),rowComparator);
        //传入filter
        scan.setFilter(filter);
        //执行scan
        ResultScanner resultScanner = table.getScanner(scan);
        Iterator<Result> iterator = resultScanner.iterator();
        List<CellData[]> retList = new ArrayList<>();
        //迭代，在迭代时，ResultScanner会发送get请求
        while(iterator.hasNext()){
            Result result=iterator.next();
            //获取rowkey
            String rowkey=Bytes.toString(result.getRow());
            //获取所有单元格
            List<Cell> list=result.listCells();
            CellData[] arr = new CellData[list.size()];
            int i = 0;
            for(Cell item:list){
                String familyOfDb=Bytes.toString(CellUtil.cloneFamily(item));
                String qualifier=Bytes.toString(CellUtil.cloneQualifier(item));
                String valueDb=Bytes.toString(CellUtil.cloneValue(item));
                CellData cellData = new CellData(familyOfDb,qualifier,rowkey,item.getTimestamp(),valueDb);
                arr[i] = cellData ;
                i ++ ;
            }
            retList.add(arr);
        }
        //用完要及时关闭，在迭代时，ResultScanner会发送get请求
        resultScanner.close();
        table.close();
        return retList;
    }

    /**
     * @param tableName
     * @param columnFamily1    第一个条件的列族名
     * @param column1          第一个条件的列名
     * @param op1              第一个条件的操作符
     * @param value1           第一个条件的值
     * @param filterIfMissing1 第一个条件的filterIfMissing，true 表示没有列值，认为不满足条件；false，没有列值，也满足条件
     * @param columnFamily2    第二个条件的列族名
     * @param column2          第二个条件的列名
     * @param op2              第二个条件的操作符
     * @param value2           第二个条件的值
     * @param filterIfMissing2 第二个条件的filterIfMissing，true 表示没有列值，认为不满足条件；false，没有列值，也满足条件
     * @return
     * @throws IOException
     */
    public List<CellData[]> filter(String tableName,
                                   String columnFamily1, String column1, String op1, String value1, boolean filterIfMissing1,
                                   String columnFamily2, String column2, String op2, String value2, boolean filterIfMissing2) throws IOException {
        Table table = conn.getTable(TableName.valueOf(tableName));
        //创建Scan
        Scan scan = new Scan();
        SingleColumnValueFilter startFilter = new SingleColumnValueFilter(
                Bytes.toBytes(columnFamily1),
                Bytes.toBytes(column1),
                getCompareOperator(op1),
                new BinaryComparator(Bytes.toBytes(value1))); // 注意是字符串
        startFilter.setFilterIfMissing(filterIfMissing1);
        SingleColumnValueFilter endFilter = new SingleColumnValueFilter(
                Bytes.toBytes(columnFamily2),
                Bytes.toBytes(column2),
                getCompareOperator(op2),
                new BinaryComparator(Bytes.toBytes(value2))); // 注意是字符串
        endFilter.setFilterIfMissing(filterIfMissing2);
        //构建过滤list，MUST_PASS_ALL 相当于 sql 中的 and
        FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ALL, startFilter, endFilter);
        //传入FilterList
        scan.setFilter(fl);
        //执行scan
        ResultScanner resultScanner = table.getScanner(scan);
        Iterator<Result> iterator = resultScanner.iterator();
        List<CellData[]> retList = new ArrayList<>();
        //迭代，在迭代时，ResultScanner会发送get请求
        while (iterator.hasNext()) {
            Result result = iterator.next();
            //获取rowkey
            String rowkey = Bytes.toString(result.getRow());
            //获取所有单元格
            List<Cell> list = result.listCells();
            CellData[] arr = new CellData[list.size()];
            int i = 0;
            for (Cell item : list) {
                String family = Bytes.toString(CellUtil.cloneFamily(item));
                String qualifier = Bytes.toString(CellUtil.cloneQualifier(item));
                String valueDb = Bytes.toString(CellUtil.cloneValue(item));
                CellData cellData = new CellData(family, qualifier, rowkey, item.getTimestamp(), valueDb);
                arr[i] = cellData;
                i++;
            }
            retList.add(arr);
        }
        //用完要及时关闭，在迭代时，ResultScanner会发送get请求
        resultScanner.close();
        table.close();
        return retList;
    }

    // 传入 =，>，>=，<，<=，!=，返回相对应的 CompareOperator
    CompareOperator getCompareOperator(String op){
        CompareOperator co = null;
        switch (op) {
            case "=":
                co = CompareOperator.EQUAL;
                break;
            case ">":
                co = CompareOperator.GREATER;
                break;
            case ">=":
                co = CompareOperator.GREATER_OR_EQUAL;
                break;
            case "<":
                co = CompareOperator.LESS;
                break;
            case "<=":
                co = CompareOperator.LESS_OR_EQUAL;
                break;
            case "!=":
                co = CompareOperator.NOT_EQUAL;
                break;
            default:
                co = CompareOperator.NO_OP;
        }
        return co;
    }



    public static void main(String[] args) throws IOException {

        HbaseUtils to=new HbaseUtils();
        try {
            to.connect("hadoop1,hadoop2,hadoop3","2181");
            System.out.println("------------------------------------------");
            System.out.println("connect success!");
            System.out.println("------------------------------------");
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                to.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

    }


}
